package cn.whuc.spark.operator

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo_aggregateByKey_case01 {
  def main(args: Array[String]): Unit = {
    // 1 创建sparkContext
    val sc: SparkContext = new SparkContext(
      new SparkConf()
        .setMaster("local[*]")
        .setAppName(" ")
    )

    // 2 编写代码
    //Abel,DataStructure,38
    //Abel,OperatingSystem,38
    // 求学生平均分
    val datas: RDD[String] = sc.textFile("input/Data01.txt")
    val stuScore: RDD[(String, Int)] = datas.map(line => {
      val strings: Array[String] = line.split(",")
      (strings(0), strings(2).toInt)
    })

    // 学上的平均分需要 总分 和 课程数，于是0值可以设定为一个元组形式的累加器
    // acc分别是 总分 和 科目数
    val rdd2: RDD[(String, (Int, Double))] = stuScore.aggregateByKey((0, 0.0))(
      // 分区内 同学生 累加器中的总分和socre的累加 和 科目数+1
      (acc, score) => (acc._1 + score, acc._2 + 1),
      // 分区间 不同累加器的 同学生的 总分相加和科目数相加
      (acc1, acc2) => (acc1._1 + acc2._1, acc1._2 + acc2._2)
      // 此时 原RDD的结构变了
    )
    // 过滤出需要的学生，对kv的v进行处理 即可得到需要的均分
    val rdd3: RDD[(String, (Int, Double))] = rdd2.filter(_._1 == "Tom")
//    val resultRDD: RDD[(String, Double)] = rdd2
//    val resultRDD: RDD[(String, Double)] = stuScore.aggregateByKey((0, 0.0))(
//      (acc, score) => (acc._1 + score, acc._2 + 1),
//      (acc1, acc2) => (acc1._1 + acc2._1, acc1._2 + acc2._2)
//    ).mapValues {
//      case (totalSocre, count) => {
//        (totalSocre / count.toDouble)
//      }
//    }

//    resultRDD.collect().foreach(println)
    rdd3.collect().foreach(println)


    // 3 关闭上下文对象
    sc.stop()
  }
}
